﻿#include "pch.h"

std::unique_ptr<TChannelSampling> channel_sampling;

#ifdef _USRDLL
TChannelSampling *getChannelSamplingPtr()
{
    return channel_sampling.get();
}
#endif

void SpectrumDataManager::addData(int32_t channel, const TRawItem &point)
{
    std::lock_guard<std::mutex> lock(mutex_);
    data_[channel].push_back(point);
}

void SpectrumDataManager::clear()
{
    std::lock_guard<std::mutex> lock(mutex_);
    for (auto &pair : data_)
    {
        pair.second.clear();
    }
}

void SpectrumDataManager::clearChannel(int32_t chn)
{
    auto it = data_.find(chn);
    if (it != data_.end())
    {
        it->second.clear();
    }
}

std::vector<TRawItem> SpectrumDataManager::getData(int32_t channel) const
{
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = data_.find(channel);
    return it != data_.end() ? it->second : std::vector<TRawItem>();
}

int64_t TChannelSampling::getOrgHandle()
{
    return org_handle;
}
void TChannelSampling::setOrgHandle(int64_t handle)
{
    org_handle = handle;
}

bool TChannelSampling::isSampling()
{
    return state_.is_sampling;
}

bool TChannelSampling::isOrgHandleSampling(int64_t nHandle)
{
    if (isSampling() && getOrgHandle() == nHandle)
    {
        return true;
    }
    return false;
}

bool TChannelSampling::ProcessSpectrumData(int32_t InstID, int32_t detID, const std::vector<TRawItem> &data)
{
    std::lock_guard<std::mutex> lock(process_mutex_);
    return ProcessSpectrumData_(InstID, detID, data);
}
bool TChannelSampling::ProcessSpectrumData4Test(int32_t InstID, int32_t detID, const std::vector<TRawItem> &data)
{
    std::lock_guard<std::mutex> lock(process_mutex_);
    TOrgHead *pOrgHead = getOrgHeadDirect(org_handle);
    int32_t chn = getChannelID(pOrgHead->Detector, (InstID << 4) | detID);
    if (chn < 0 || chn >= C4)
    {
        STD_CERR << "Invalid channel number: " << chn << STD_ENDL;
        return false;
    }
    data_manager_.clearChannel(chn);
    return ProcessSpectrumData_(InstID, detID, data);
}
bool TChannelSampling::ProcessSpectrumData_(int32_t InstID, int32_t detID, const std::vector<TRawItem> &data)
{
    TOrgHead *pOrgHead = getOrgHeadDirect(org_handle);
    int32_t chn = getChannelID(pOrgHead->Detector, (InstID << 4) | detID);
    if (chn < 0 || chn >= C4)
    {
        STD_CERR << "Invalid channel number: " << chn << STD_ENDL;
        return false;
    }

    bool is_valid_data = false;
    int32_t i = 0;
    for (const auto &spectrumPoint : data)
    {
        if (state_.step[chn] == 0)
        { // 尚未开始记录做样数据。
            if (spectrumPoint.Status == 0)
            {
                if (i == 0)
                {
                    STD_COUT << "[" << chn << "]" << "开始做样前的基线数据..." << i << STD_ENDL;
                }
            }
            else
            { // 记录第一个采样数据。
                state_.start_time[chn] = now_ms();
                state_.gc_start_time[chn] = spectrumPoint.Time;

                data_manager_.addData(chn, spectrumPoint);
                state_.stop_time[chn] = now_ms();
                state_.gc_stop_time[chn] = spectrumPoint.Time;
                state_.gc_last_value[chn] = spectrumPoint.Data;

                state_.step[chn] = 1;
                STD_COUT << "[" << chn << "]" << "处理第一个做样数据。" << i << STD_ENDL;
            }
            is_valid_data = true;
        }
        else if (state_.step[chn] == 1)
        {
            if (spectrumPoint.Status == 1)
            { // 开始持续记录做样数据。
                data_manager_.addData(chn, spectrumPoint);
                state_.stop_time[chn] = now_ms();
                state_.gc_stop_time[chn] = spectrumPoint.Time;
                state_.gc_last_value[chn] = spectrumPoint.Data;
                if (i == 0)
                {
                    STD_COUT << "[" << chn << "]" << "持续做样数据..." << i << STD_ENDL;
                }
            }
            else
            { // 持续记录做样数据时检测到停止做样。
                STD_COUT << "[" << chn << "]" << "停止做样。" << i << STD_ENDL;
                state_.step[chn] = 2;
            }
            is_valid_data = true;
        }
        else
        {
            if (spectrumPoint.Status == 1)
            {
                if (i == 0)
                {
                    STD_CERR << "停止做样后又接受到了做样数据。" << i << STD_ENDL;
                }
            }
            else
            {
                if (i == 0)
                {
                    STD_COUT << "[" << chn << "]" << "停止做样后的基线数据..." << i << STD_ENDL;
                }
            }
            is_valid_data = false;
        }
        i++;
    }
    return is_valid_data;
}

void TChannelSampling::StartSampling()
{
    if (state_.is_sampling.exchange(true))
    {
        throw std::runtime_error("Sampling is already in progress");
    }

    for (int32_t chn = 0; chn < C4; chn++)
    {
        state_.step[chn] = 0;
        state_.start_time[chn] = 0;
        state_.stop_time[chn] = 0;
        state_.gc_start_time[chn] = 0;
        state_.gc_stop_time[chn] = 0;
        state_.gc_last_value[chn] = 0;
    }
    data_manager_.clear();
    clearError();

    sampling_thread_ = std::thread([this]()
                                   {        
        STD_COUT << "采样线程开始。" << STD_ENDL;
        try {
            ConfigINI& redisConfig = ConfigINI::GetInstance();
            std::string  redisHost = redisConfig.GetString( "RedisHost", "127.0.0.1" );
            int32_t      redisPort = redisConfig.GetInt( "RedisPort", 6379 );
            int32_t      redisConsumeTimeout = redisConfig.GetInt( "RedisConsumeTimeout", 5000 );
            RedisManager redis( redisHost, redisPort,sampling_config_.redis_timeout_msec );
            STD_COUT << "redisHost=" << redisHost << ", redisPort=" << redisPort << ", redisConsumeTimeout=" << redisConsumeTimeout << STD_ENDL;
            auto onMessage = [ this ]( auto channel, auto msg ) {
                try {
                    STD_COUT << "Received message on channel=" << channel << ", msg.size()=" << msg.size() << STD_ENDL;
                    boost::json::object joMessage = boost::json::parse( msg ).as_object();

                    int32_t     InstID = JsonUtils::get_int<int32_t>( joMessage, "InstID" );
                    std::string Sender = JsonUtils::get_string( joMessage, "Sender" );

                    if ( joMessage.contains( "Content" ) && joMessage[ "Content" ].is_object() ) {
                        auto    joContent = joMessage[ "Content" ].as_object();
                        int32_t detID = JsonUtils::get_int<int32_t>( joContent, "detID" );
                        int32_t length = JsonUtils::get_int<int32_t>( joContent, "length" );

                        std::vector<TRawItem> spectrumData;
                        if ( joContent.contains( "data" ) && joContent[ "data" ].is_array() ) {
                            auto dataArray = joContent[ "data" ].as_array();
                            if ( dataArray.size() != static_cast<size_t>( length ) ) {
                                STD_CERR << "Data length mismatch: expected " << length << ", got " << dataArray.size() << STD_ENDL;
                            }
                            else {
                                STD_COUT << "Data Received: detID=" << detID << ", dataArray.size()=" << dataArray.size() << STD_ENDL;
                            }

                            for ( const auto& pointValue : dataArray ) {
                                if ( pointValue.is_object() ) {
                                    auto     joPoint = pointValue.as_object();
                                    TRawItem point;
                                    point.Status = JsonUtils::get_int<int16_t>( joPoint, "wStatus" );
                                    point.Time = JsonUtils::get_int<int32_t>( joPoint, "iTime" );
                                    point.Data = JsonUtils::get_int<int32_t>( joPoint, "iData" );
                                    spectrumData.push_back( point );
                                }
                            }
                            STD_COUT << "ProcessSpectrumData InstID=" << InstID << ", detID=" << detID << STD_ENDL;
                            ProcessSpectrumData( InstID, detID, spectrumData );
                        }
                    }
                } catch ( const std::exception& e ) {
                    STD_CERR << "Error processing message: " << e.what() << STD_ENDL;
                }
            };

            redis.subscribe( "SpectrumData", onMessage );

            while ( state_.is_sampling ) {
                if(!redis.consume()){
                    setError( ERR_CODE_REDIS_TIMEOUT, ERR_MESSAGE_REDIS_TIMEOUT );
                    state_.is_sampling = false;
                    break;
                }
                std::this_thread::sleep_for( 100ms );
            }
        } catch ( const std::exception& e ) {
            STD_CERR << "Sampling thread error: " << e.what() << STD_ENDL;
            state_.is_sampling = false;
        } 
        STD_COUT << "采样线程结束了。" << STD_ENDL; });

    // 启动采样后预期 start_timeout_msec 毫秒内数据状态应切变为1
    auto start_wait_until = now_ms() + sampling_config_.start_timeout_msec;

    STD_COUT << "开始采样，等待数据状态切为1..." << STD_ENDL;
    TOrgHead *pOrgHead = getOrgHeadDirect(org_handle);
    while (state_.is_sampling)
    {
        if (now_ms() > start_wait_until)
        {
            setError(ERR_CODE_START_TIMEOUT, ERR_MESSAGE_START_TIMEOUT);
            state_.is_sampling = false;
            STD_COUT << "超时数据状态没有切为1，开始采样失败。" << STD_ENDL;
            break;
        }

        bool toggled = true;
        for (int32_t chn = 0; chn < C4; chn++)
        {
            if (isValidChannel(pOrgHead->Detector, chn))
            {
                if (state_.step[chn] <= 0)
                {
                    toggled = false;
                }
            }
        }
        if (toggled)
        {
            STD_COUT << "数据状态切为1。" << STD_ENDL;
            break;
        }
        std::this_thread::sleep_for(100ms);
    }

    // 失败后等待子线程结束
    if (!state_.is_sampling)
    {
        STD_COUT << "采样失败，判定采样线程..." << STD_ENDL;
        if (sampling_thread_.joinable())
        {
            STD_COUT << "采样失败，等待采样线程终止..." << STD_ENDL;
            sampling_thread_.join();
        }
        STD_COUT << "采样失败，采样线程已经终止。" << STD_ENDL;
    }
}

bool TChannelSampling::StopSampling()
{
    TOrgHead *pOrgHead = getOrgHeadDirect(org_handle);
    if (!state_.is_sampling.exchange(false))
    {
        setError(ERR_CODE_CHANNEL_IS_NOT_SAMPLING, ERR_MESSAGE_CHANNEL_IS_NOT_SAMPLING);
    }

    STD_COUT << "判定采样线程..." << STD_ENDL;
    if (sampling_thread_.joinable())
    {
        STD_COUT << "等待采样线程终止..." << STD_ENDL;
        sampling_thread_.join();
    }
    STD_COUT << "采样线程已经终止。" << STD_ENDL;

    if (hasError())
    {
        SET_ERROR(getLastErrorCode(), getLastErrorMessage().c_str());
        return false;
    }

    for (int32_t chn = 0; chn < C4; chn++)
    {
        if (isValidChannel(pOrgHead->Detector, chn))
        {
            // 检查时间同步
            // 当前时间与停止采样时间的误差应在 stop_time_sync_msec 毫秒内
            if (abs((int64_t)(now_ms() - state_.stop_time[chn])) > sampling_config_.stop_time_sync_msec)
            {
                setError(ERR_CODE_STOP_TIME_NOT_SYNC, ERR_MESSAGE_STOP_TIME_NOT_SYNC);
            }

            // 采样时长应保持与Sample.Minutes一致，采样时长可以小于Minutes，但不能大于Minutes。
            if (state_.stop_time[chn] - state_.start_time[chn] - floor(0.01 * pOrgHead->Sample[chn].Minutes * 60 * 1000) > sampling_config_.gatcher_time_sync_msec)
            {
                STD_CERR << ERR_MESSAGE_GATHER_TIME_NOT_SYNC << ": Delta=" << state_.stop_time[chn] - state_.start_time[chn] << ", Minutes=" << pOrgHead->Sample[chn].Minutes << STD_ENDL;
                setError(ERR_CODE_GATHER_TIME_NOT_SYNC, ERR_MESSAGE_GATHER_TIME_NOT_SYNC);
            }
        }
    }

    // 保存数据到OrgFile
    if (getLastErrorCode() == 0)
    {
        STD_COUT << "采样成功，保存谱图数据到句柄空间 。" << STD_ENDL;
        saveDataToOrgFile();
    }
    SET_ERROR(getLastErrorCode(), getLastErrorMessage().c_str());
    return !hasError();
}

void TChannelSampling::saveDataToOrgFile()
{
    std::lock_guard<std::mutex> lock(org_mutex_);
    TOrgHead *pOrgHead = getOrgHeadDirect(org_handle);
    for (int32_t ch = 0; ch < C4; ++ch)
    {
        if (isValidChannel(pOrgHead->Detector, ch))
        {
            auto RawData = data_manager_.getData(ch);
            int32_t RawDataCount = (int32_t)RawData.size();

            // 通道采集点数应与采集时长数据同步
            int32_t msecondsFromData = std::floor(1000.0 * RawDataCount / (pOrgHead->Frequency != 0 ? pOrgHead->Frequency : 1));
            int32_t msecondsFromTime = state_.stop_time[ch] - state_.start_time[ch];
            if (abs(msecondsFromData - msecondsFromTime) > sampling_config_.gatcher_time_sync_msec)
            {
                STD_CERR << ERR_MESSAGE_GATHER_TIME_NOT_SYNC << ": RawDataCount=" << RawDataCount << STD_ENDL;
                STD_CERR << ERR_MESSAGE_GATHER_TIME_NOT_SYNC << ": pOrgHead->Frequency=" << pOrgHead->Frequency << STD_ENDL;
                STD_CERR << ERR_MESSAGE_GATHER_TIME_NOT_SYNC << ": msecondsFromData=" << msecondsFromData << STD_ENDL;
                STD_CERR << ERR_MESSAGE_GATHER_TIME_NOT_SYNC << ": msecondsFromTime=" << msecondsFromTime << STD_ENDL;
                setError(ERR_CODE_GATHER_TIME_NOT_SYNC, ERR_MESSAGE_GATHER_TIME_NOT_SYNC);
            }

            // 采样过程中调用谱图相关接口函数会失败，因此调用setOrgRawData前设置org_handle=-1避免采样过程不允许变更谱图数据的失败因素。
            int64_t saved_handle = org_handle;
            org_handle = -1;
#ifdef _USRDLL            
            setOrgRawData_(saved_handle, ch, RawData.data(), RawDataCount);
#else
            setOrgRawData(saved_handle, ch, RawData.data(), RawDataCount);
#endif            
            org_handle = saved_handle;

            pOrgHead->Inject[ch].StartTime = (int32_t)(state_.start_time[ch] / 1000);
            pOrgHead->Inject[ch].StopTime = (int32_t)(state_.stop_time[ch] / 1000);
        }
    }
}

TChannelSampling::~TChannelSampling()
{
    if (state_.is_sampling)
    {
        StopSampling();
    }
}

bool TChannelSampling::hasError()
{
    return state_.last_error_code != 0;
}
void TChannelSampling::clearError()
{
    setError(0, "");
}
void TChannelSampling::setError(int32_t code, std::string message)
{
    std::lock_guard<std::mutex> lock(error_mutex_);
    state_.last_error_code = code;
    state_.last_error_message = message;
}
int32_t TChannelSampling::getLastErrorCode() const
{
    std::lock_guard<std::mutex> lock(error_mutex_);
    return state_.last_error_code;
}

std::string TChannelSampling::getLastErrorMessage() const
{
    std::lock_guard<std::mutex> lock(error_mutex_);
    return state_.last_error_message;
}
